package com.shujia.rec.sink

import com.shujia.rec.common.Constants
import com.shujia.rec.entry.CaseClass.CountEntry
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import redis.clients.jedis.Jedis

class RedisSink extends RichSinkFunction[List[CountEntry]] {

  var jedis: Jedis = _

  override def open(parameters: Configuration): Unit = {
    jedis = new Jedis(Constants.REDIS_HOST, 6379)

  }

  override def invoke(value: List[CountEntry], context: SinkFunction.Context[_]): Unit = {

    jedis.del("rec:topn")
    value.foreach(log => {
      val proid = log.proId
      val count = log.count
      val time = log.end
      val line = s"$proid,$count,$time"
      jedis.lpush("rec:topn", line)
    })
  }

  override def close(): Unit = {
    jedis.close()
  }
}
